package ar.edu.itba.pod.legajo45002;

import java.rmi.RemoteException;
import java.rmi.registry.Registry;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;


import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;

import ar.edu.itba.pod.legajo45002.api.NodeStats;
import ar.edu.itba.pod.legajo45002.api.Result;
import ar.edu.itba.pod.legajo45002.api.Result.Item;
import ar.edu.itba.pod.legajo45002.api.SPNode;
import ar.edu.itba.pod.legajo45002.api.Signal;
import ar.edu.itba.pod.legajo45002.api.SignalProcessor;
import ar.edu.itba.pod.legajo45002.messageObjects.GenericMessage;
import ar.edu.itba.pod.legajo45002.messageObjects.ResultObject;
import ar.edu.itba.pod.legajo45002.messageObjects.SignalListObject;
import ar.edu.itba.pod.legajo45002.messageObjects.SignalObject;

public class ProcessingNode extends ReceiverAdapter implements SignalProcessor, SPNode {
	private  ConcurrentHashMap<Address, List<DistributedSignal>> signalsMap = new ConcurrentHashMap<>();
	private ConcurrentHashMap<Address, List<DistributedSignal>> backupSignalsMap = new ConcurrentHashMap<>();

	private BlockingQueue<Signal> singleNodeSignals = new LinkedBlockingDeque<>();

	private BlockingQueue<CompSignals> signalsToProcess = new LinkedBlockingDeque<>();
	private BlockingQueue<Result> threadResultsList = new LinkedBlockingDeque<>();

	private BlockingQueue<Integer> findSimilarToTasks = new LinkedBlockingDeque<>();

	private BlockingQueue<Message>recievedMessagesQueue;
	private BlockingQueue<Signal> addSignalQueue;
	private BlockingQueue<View>viewAcceptedQueue;

	private ExecutorService executors;
	private ExecutorService generalExecutors;

	private BlockingQueue<Result> resultsQueue = new LinkedBlockingDeque<Result>();

	private AtomicBoolean isNewNode = new AtomicBoolean(true);
	private AtomicBoolean degraded = new AtomicBoolean(true);
	private AtomicBoolean taskInterrupted = new AtomicBoolean(false);
	private AtomicBoolean addInterruptedBoolean = new AtomicBoolean();

	private AddSignalToClusterThread addSignalToClusterThread;
	private ReceivedMessageThread receivedMessageThread;
	private ViewAcceptedThread viewAcceptedThread;
	private int processorTrheadsCount;
	private Address deathSuspectedNode;
	private String cluster = null;
	private int receivedSignals = 0;
	private AtomicBoolean interruptedProccessing = new AtomicBoolean();
	private Thread findSimilarInvocationThread;

	private Address Coordinator;

	private AtomicBoolean receiveThreadFinished = new AtomicBoolean();

	private AtomicBoolean iStartedFind = new AtomicBoolean(false);

	private CountDownLatch countLatch;
	private CountDownLatch notDegradedAnyMore;
	private CountDownLatch newNodeSignalsLatch;

	//	private Semaphore findSimilarSemaphore = new Semaphore(1);

	JChannel channel = null;

	CountDownLatch redistributeLocalLatch;

	private View lastView;

	private AtomicInteger clusterSize = new AtomicInteger();

	public ProcessingNode(int processingThreads) {
		executors = Executors.newFixedThreadPool(processingThreads);
		processorTrheadsCount = processingThreads;
	}


	public synchronized void setLastView(View lastView) {
		this.lastView = lastView;
	}

	public synchronized View getLastView() {
		return lastView;
	}


	/*
	 * SPNode methods
	 * 
	 */
	@Override
	public void join(String clusterName) throws RemoteException {
		System.out.println( "joining cluster "+ clusterName);
		if (cluster != null) {
			throw new IllegalStateException("Already in cluster " + cluster);
		}
		if (!singleNodeSignals.isEmpty()) {
			throw new IllegalStateException("Can't join a cluster because there are signals already stored");
		}

		interruptedProccessing.set(false);
		this.cluster = clusterName;
		receiveThreadFinished.set(false);
		addInterruptedBoolean.set(false);

		signalsMap = new ConcurrentHashMap<>();
		backupSignalsMap = new ConcurrentHashMap<>();

		findSimilarToTasks = new LinkedBlockingDeque<>();
		recievedMessagesQueue = new LinkedBlockingDeque<>();
		viewAcceptedQueue = new LinkedBlockingDeque<>();
		addSignalQueue = new LinkedBlockingDeque<>();

		receivedMessageThread = new ReceivedMessageThread();
		viewAcceptedThread = new ViewAcceptedThread();
		addSignalToClusterThread = new AddSignalToClusterThread();

		generalExecutors = Executors.newFixedThreadPool(3);

		generalExecutors.execute(receivedMessageThread);
		generalExecutors.execute(viewAcceptedThread);
		generalExecutors.execute(addSignalToClusterThread);

		isNewNode.set(true);

		try {
			channel=new JChannel();
			channel.setReceiver(this);

		} catch (Exception e) {
			e.printStackTrace();
		}
		try {
			channel.connect(clusterName);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void exit() throws RemoteException {
		System.out.println("Exiting cluster");
		singleNodeSignals = new LinkedBlockingDeque<Signal>();
		backupSignalsMap = new ConcurrentHashMap<>();
		cluster = null;
		if( channel != null){
			receiveThreadFinished.set(true);
			addInterruptedBoolean.set(true);
			generalExecutors.shutdownNow();
			channel.close();
			channel = null;
		}
	}


	@Override
	public NodeStats getStats() throws RemoteException {

		return new NodeStats(
				cluster == null ? "standalone" : "cluster: " + cluster + "node: " + channel == null? "standalone" : channel.getAddressAsString(), 
						receivedSignals,
						cluster == null ? singleNodeSignals.size() : signalsSize(signalsMap), 
								signalsSize(backupSignalsMap), 
								degraded.get() || clusterSize.get() <= 1 );
	}

	private int channelSize(){
		return clusterSize.get();
	}
	/*
	 * SignalProcessor methods
	 * 
	 */
	private class AddSignalToClusterThread implements Runnable{
		@Override
		public void run() {
			while (!addInterruptedBoolean.get()){
				Signal signal = null;
				try {
					signal = addSignalQueue.take();
					DistributedSignal dsig;

					Address destinyNode = getRandomNode();
					Address backupNode = getRandomNodeExceptNode(destinyNode);
					if( backupNode == null){
						backupNode = destinyNode;
					}
					dsig = new DistributedSignal(signal, destinyNode, backupNode);
					sendSignalToNode(MessageNames.ADD_SIGNAL_TO_MY_LIST, destinyNode, dsig);	
					if( channelSize() >= 2){
						sendSignalToNode(MessageNames.ADD_SIGNAL_TO_BACKUP, backupNode, dsig);	
					}
				} catch (InterruptedException e1) {
					addInterruptedBoolean.set(true);
					//	e1.printStackTrace();
				}
			}
		}
	}

	@Override
	public void add(Signal signal) throws RemoteException {
		if( !iamAlone() ){
			try {
				addSignalQueue.put(signal);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}else{
			try {
				singleNodeSignals.put(signal);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}


	private boolean iamAlone(){
		return channel == null ;
	}

	private boolean clusterDegraded(){
		return cluster != null && clusterSize.get() > 1 && degraded.get();
	}


	private void sendSignalToNode(MessageNames msg,  Address node, DistributedSignal s){
		GenericMessage message = new SignalObject(msg,  s);
		try {
			if( channel != null )
				channel.send(node, message);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private class findSimilarThread extends Thread{
		private Signal signalToProcess;

		public findSimilarThread(Signal signal) {
			this.signalToProcess = signal;
			findSimilarToTasks = new LinkedBlockingDeque<>();
			try {
				findSimilarToTasks.put(1);
			} catch (InterruptedException e2) {
				e2.printStackTrace();
			}
		}

		@Override
		public void run() {

			while( interruptedProccessing.get() ){
				try {
					findSimilarToTasks.take();	
					taskInterrupted.set(false);
					System.out.println("Taking task from Queue...");
					resultsQueue = new LinkedBlockingDeque<>();
					countLatch = new CountDownLatch(channel.getView().getMembers().size());
					BlockingQueue<Signal> addedSignals = new LinkedBlockingDeque();

					addedSignals.addAll(addSignalQueue);
					for( Signal s: addedSignals){
						signalsToProcess.put(new CompSignals(s, signalToProcess));
					}

					DistributedSignal s = new DistributedSignal(signalToProcess, null, null);
					System.out.println("Sending process Signal...");
					sendSignalToNode(MessageNames.PROCESS_SIGNAL,null,  s);

					countLatch.await(3000 + signalsSize(signalsMap) * 2 + signalsToProcess.size(), TimeUnit.MILLISECONDS);
					interruptedProccessing.set(false);
				} catch (InterruptedException e) {
					executors.shutdownNow();
					System.out.println(" await interrpuped ");
					System.out.println("Bloqued till cluster quits degradation.. ");			
				}
			}
		}
	}

	@Override
	public Result findSimilarTo(Signal signal) throws RemoteException {
		if (signal == null) {
			throw new IllegalArgumentException("Signal cannot be null");
		}
		receivedSignals++;
		iStartedFind.set(true);
		executors = Executors.newFixedThreadPool(processorTrheadsCount);

		if( !iamAlone() ){
			interruptedProccessing.set(true);
			findSimilarInvocationThread = new findSimilarThread(signal);
			findSimilarInvocationThread.start();
			try {
				findSimilarInvocationThread.join();
				iStartedFind.set(false);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}else{
			signalsToProcess = new LinkedBlockingDeque<>( );
			return procesSignsl(signal);
		}

		Result mainResult = new Result(signal);
		for( Result r : resultsQueue){
			for( Item i:  r.items() ){
				mainResult = mainResult.include(i);
			}
		}
		System.out.println("mainResult: " + mainResult);
		System.out.println("all nodes returned");
		return mainResult;
	}



	/*
	 * RecieveAdapter methods
	 * 
	 */
	@Override
	public void viewAccepted(View new_view) {
		System.out.println("VIEW ACCEPTED...");
		clusterSize.set(new_view.getMembers().size());
		Coordinator = new_view.getMembers().get(0);
		try {
			System.out.println("Putting new View in queue..");
			viewAcceptedQueue.put(new_view);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}


	private void doRedistributionProcess( Address deathNode){
		System.out.println("doRedistributionProcess...");

		if(clusterSize.get() == 1){
			redistributeSingleNode(deathNode);
		}else{
			redistributeDeathNode( deathNode );
		}

		if( iStartedFind.get()){
			findSimilarInvocationThread.interrupt();
		}
		if( iAmTheCoordinator() ){
			try {
				sendMessage(null, MessageNames.SET_ALL_DEGRADED );

				notDegradedAnyMore.await(10000 + signalsSize(signalsMap) * 2, TimeUnit.MILLISECONDS);
				System.out.println("---- EVERYONE FINISHED DISTRIBUTING -----");
				//	System.out.println("SIGNALS: "+ signalsMap);
				//	System.out.println("BACKUP: "+ backupSignalsMap);
				sendMessage(null, MessageNames.WE_ARE_ALL_OK);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	private boolean iAmTheCoordinator() {
		return Coordinator.equals(channel.getAddress());
	}

	private void redistributeSingleNode(Address deathSuspectedNode) {
		System.out.println("--- redistributeSingleNode() ---- " + deathSuspectedNode);
		//		System.out.println("SIGNALS: "+ signalsMap);
		//		System.out.println("BACKUP: "+ backupSignalsMap);
		List<DistributedSignal> backs = MapToList(backupSignalsMap);
		List<DistributedSignal> l = signalsMap.get(deathSuspectedNode);
		if( l != null){
			signalsMap.put(channel.getAddress(), l);
			for( DistributedSignal s: backs){
				s.setWhoHasIt(channel.getAddress());
				s.setWhoBacksItUp(channel.getAddress());
				addSignalToMyList(s);
			}
		}
		signalsMap.remove(deathSuspectedNode);
		backupSignalsMap = new ConcurrentHashMap<>(); 

		sendMessage(Coordinator, MessageNames.FINISHED_DISTRIBUTION_OF_SIGNALS);

		//	System.out.println("SIGNALS: "+ signalsMap);
		//	System.out.println("BACKUP: "+ backupSignalsMap);
	}


	private synchronized void populateMaps (Map<Address, List<DistributedSignal>> distmap, Map<Address, List<DistributedSignal>> backupMap,List<DistributedSignal> signalsToDistribute ){
		List<DistributedSignal> l;

		for( DistributedSignal s : signalsToDistribute){
			Address destinyNode = getRandomNode();
			Address backUpNode = getRandomNodeExceptNode(destinyNode);
			s.setWhoBacksItUp(backUpNode);
			s.setWhoHasIt(destinyNode);
			if( distmap.get(destinyNode) == null) {
				l = new LinkedList<>();
				l.add(s);
				distmap.put(destinyNode, l);
			}else{
				distmap.get(destinyNode).add(s);
			}

			if( backupMap.get(backUpNode) == null) {
				l = new LinkedList<>();
				l.add(s);
				backupMap.put(backUpNode, l);
			}else{
				backupMap.get(backUpNode).add(s);
			}
		}
	}

	private  void redistributeDeathNode(final Address deathnode) {
		
		Thread distributeThread = new Thread(){
			
			public void run() {
				System.out.println("--- redistributeDeathNode() ----");

				Map<Address, List<DistributedSignal>> distmap = new HashMap<>();
				Map<Address, List<DistributedSignal>> backupMap = new HashMap<>();

				
				if( signalsMap.get(deathnode) != null){
					populateMaps( distmap, backupMap, signalsMap.get(deathnode));
				}
				if( backupSignalsMap.get(deathnode) != null){
					populateMaps(distmap, backupMap, backupSignalsMap.get(deathnode));
				}

				signalsMap.remove(deathnode);
				backupSignalsMap.remove(deathnode);
				
				redistributeLocalLatch = new CountDownLatch(distmap.size() + backupMap.size());

				for( Address destinyNode : distmap.keySet()){
					System.out.println("Sending signals "+ distmap.get(destinyNode).size() + " to node "+ destinyNode);
					sendSignalsListToNode( MessageNames.SEND_SIGNALS_TO_NODE, destinyNode, distmap.get(destinyNode));
				}

				for( Address destinyNode : backupMap.keySet()){
					System.out.println("Sending backups "+ backupMap.get(destinyNode).size() + " to node "+ destinyNode);
					sendSignalsListToNode( MessageNames.SEND_SIGNALS_TO_NODES_BACKUP, destinyNode, backupMap.get(destinyNode));
				}

			

				try {
					redistributeLocalLatch.await( 1000 + signalsSize(signalsMap), TimeUnit.MILLISECONDS);

					sendMessage(Coordinator, MessageNames.FINISHED_DISTRIBUTION_OF_SIGNALS);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			};
		};
		
		distributeThread.start();
	}


	private void sendSignalsListToNode(MessageNames msg,  Address node, List<DistributedSignal> list){
		GenericMessage message = new SignalListObject(msg,  list);
		try {
			if( channel != null )
				channel.send(node, message);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}




	private void sendMessage(Address node, MessageNames msg) {
		GenericMessage message = new GenericMessage(msg);
		try {
			if( channel != null)
				channel.send(node, message );
		} catch (Exception e) {
			e.printStackTrace();
		}		
	}


	private synchronized int signalsSize( ConcurrentHashMap<Address, List<DistributedSignal>> map){
		int size = 0;
		for( List<DistributedSignal> l : map.values()){
			size += l.size();
		}
		return size;
	}


	private synchronized void removeSignalFromMyList(Message msg) {
		//	System.out.println("----- removeSignalFromMyList() ------");
		SignalObject so = (SignalObject) msg.getObject();	
		List<DistributedSignal> l;
		if( clusterSize.get() == 2){
			l = signalsMap.get(so.getDistributedSignal().getWhoHasIt());
		}else{
			l = signalsMap.get(so.getDistributedSignal().getWhoBacksItUp());
		}
		if( l != null){
			l.remove(so.getDistributedSignal());
		}
	}

	private void giveSignalsTwoNodeCluster( Address node ){

		System.out.println("--- giving signals to node: "+node +" ----");
		System.out.println("we are 2 nodes");

		List<DistributedSignal> sigs = signalsMap.get(channel.getAddress());
		if( sigs != null ){
			signalsMap.put(node, sigs);
			signalsMap.remove(channel.getAddress());
		}

		List<DistributedSignal> l = MapToList(signalsMap);
		List <DistributedSignal> newl = new LinkedList<>(l.subList(0, l.size() / 2));
		List<DistributedSignal> backupsToSend =  new LinkedList<> ( l.subList(l.size() / 2, l.size()));
		System.out.println("Transfering " +newl.size() + " to node " + node);

		for( DistributedSignal s : newl){
			s.setWhoBacksItUp(channel.getAddress());
			s.setWhoHasIt(node);
			signalsMap.get(node).remove(s);
		}
		sendSignalsListToNode(MessageNames.SEND_SIGNALS_TO_NODES_BACKUP, channel.getAddress(), newl);
		sendSignalsListToNode(MessageNames.SEND_SIGNALS_TO_NEW_NODE, node, newl);

		for(DistributedSignal bk : backupsToSend){
			bk.setWhoHasIt(channel.getAddress());
			bk.setWhoBacksItUp(node);
		}
		sendSignalsListToNode(MessageNames.SEND_SIGNALS_TO_NODES_BACKUP, node, backupsToSend);
	}


	private synchronized void giveSignalsToBigCluster(Address node ){
		System.out.println("we are more than 2 nodes");
		System.out.println("sending to " + node);

		int signalsSize = signalsSize(signalsMap);
		int signalsIShouldHave = signalsSize * (  clusterSize.get() - 1 ) / clusterSize.get();

		System.out.println("Transfering " + (signalsSize - signalsIShouldHave ) + " to node " + node);

		List<DistributedSignal> signalsToSend = new LinkedList<>();
		List<DistributedSignal> l = MapToList(signalsMap).subList(0,  signalsSize - signalsIShouldHave);
		signalsToSend.addAll(l);
		
		for( DistributedSignal s: signalsToSend  ){
			s.setWhoHasIt(node);
			signalsMap.get(s.getWhoBacksItUp()).remove(s);
		}
		
		distributeSignals(node, signalsToSend);
	}

	public void giveSignalsToNode(final Address node){

		if(!node.equals(channel.getAddress()) ){
			if( channelSize() == 2 ){
				giveSignalsTwoNodeCluster(node);
			}else{
				giveSignalsToBigCluster(node);
			}
		}
	}

	private synchronized List<DistributedSignal> MapToList(ConcurrentHashMap<Address, List<DistributedSignal>> signals){
		List<DistributedSignal> allSignals = new LinkedList<>();
		for( List<DistributedSignal> list : signals.values()){
			allSignals.addAll(list);
		}
		return allSignals;
	}


	private  void  distributeSignals( Address node,List<DistributedSignal> signals){
		//	System.out.println("SIGNALS: "+ signalsMap);
		//	System.out.println("BACKUP: "+ backupSignalsMap);
		sendSignalsListToNode(MessageNames.SEND_SIGNALS_TO_NEW_NODE, node, signals);

		for( DistributedSignal s : signals){
			s.setWhoHasIt(node);
			sendSignalToNode(MessageNames.MODIFY_BACKEDUP_SIGNAL, s.getWhoBacksItUp(), s);
		}		
	}

	@Override
	public void suspect(Address mbr){
		System.out.println("crashed suspected node: " +mbr);
		deathSuspectedNode = mbr;
	}


	@Override
	public void receive(Message msg) {

		try {
			recievedMessagesQueue.put(msg);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	private void addTaskToFindQueue(){
		if( iStartedFind.get() ){
			System.out.println("---- RE STARTING FIND SIMILAR -----");

			try {
				findSimilarToTasks.put(1);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}



	private synchronized void notifyBackUpChanged(Message msg) {
		SignalObject sig = (SignalObject)msg.getObject();
		DistributedSignal distributedSignal = sig.getDistributedSignal();
		if( backupSignalsMap.containsKey( distributedSignal.getWhoHasIt())){
			backupSignalsMap.get( distributedSignal.getWhoHasIt() ).add(distributedSignal);
		}else{
			List<DistributedSignal> newList =  new LinkedList<DistributedSignal>();
			newList.add(distributedSignal);
			backupSignalsMap.put(distributedSignal.getWhoHasIt(),newList );
		}
		if( backupSignalsMap.containsKey( msg.getSrc())){
			backupSignalsMap.get( msg.getSrc()).remove(distributedSignal);
		}
	}



	private synchronized void storeBackUpSignal(Message msg) {
		SignalObject sig = (SignalObject)msg.getObject();
		DistributedSignal distributedSignal = sig.getDistributedSignal();
		//	System.out.println("who has this signal: "+ distributedSignal.getWhoHasIt());
		if( backupSignalsMap.containsKey(distributedSignal.getWhoHasIt())){
			backupSignalsMap.get(distributedSignal.getWhoHasIt()).add(distributedSignal);
		}else{
			List<DistributedSignal> newList =  new LinkedList<DistributedSignal>();
			newList.add(distributedSignal);
			backupSignalsMap.put(distributedSignal.getWhoHasIt(),newList );
		}
	}



	private Address getRandomNode() {
		int rand = (int)(Math.random() * channel.getView().getMembers().size());
		return channel.getView().getMembers().get(rand);
	}

	private Address getRandomNodeExceptNode( Address node){
		List<Address> l = new LinkedList<>( channel.getView().getMembers());
		l.remove(node);
		int rand = (int)(Math.random() * l.size());
		if( l.isEmpty()){
			return null;
		}
		return l.get(rand);
	}

	public synchronized void addSignalToMyList(DistributedSignal distributedSignal){
		//		System.out.println("--- Adding signal to my list ---");
		if( signalsMap.containsKey(distributedSignal.getWhoBacksItUp())){
			signalsMap.get(distributedSignal.getWhoBacksItUp()).add(distributedSignal);
		}else{
			List<DistributedSignal> newList =  new LinkedList<DistributedSignal>();
			newList.add(distributedSignal);
			signalsMap.put(distributedSignal.getWhoBacksItUp(),newList );
		}
	}



	public void doProcessWork( final Message msg){
		final SignalObject s = (SignalObject) msg.getObject();

		taskInterrupted.set(false);
		System.out.println("started processing...");
		Result localMainResult = procesSignsl(s.getDistributedSignal().getS());
		if(!taskInterrupted.get()){
			System.out.println("ended processing...");

			try {
				ResultObject r = new ResultObject(MessageNames.RESULTS_RETURNED, localMainResult);
				System.out.println("Sending results...");
				channel.send(msg.getSrc(), r);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}else{
			System.out.println("Processing cancelled..");
		}
	}


	public Result procesSignsl( final Signal sig){
		executors = Executors.newFixedThreadPool(processorTrheadsCount);
		threadResultsList = new LinkedBlockingDeque<>();

		if( iamAlone()){
			for( Signal s : singleNodeSignals){
				try {
					signalsToProcess.put(new CompSignals(s, sig));
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}else{

			for( DistributedSignal s : MapToList(signalsMap)){
				try {
					//		System.out.println(s);
					signalsToProcess.put(new CompSignals(s.getS(), sig));
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		int size = signalsToProcess.size();
		System.out.println("Signals to process: " +size);

		try {
			for( int i = 0; i < processorTrheadsCount; i++){
				executors.execute(new ProccessingEngine( sig));
			}

			executors.shutdown();

			executors.awaitTermination(size * 2  , TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

		Result localMainResult = new Result(sig);
		for( Result r : threadResultsList){
			for( Item i: r.items()){			
				localMainResult = localMainResult.include(i);
			}
		}
		System.out.println("localResult: "+localMainResult);
		return localMainResult;
	}


	private class ProccessingEngine implements Runnable{
		private Signal sig;
		public ProccessingEngine(Signal s) {
			this.sig = s;
		}
		@Override
		public void run() {
			try {
				System.out.println("Thread started...");
				Result result = new Result( sig);
				CompSignals cmp;
				while(!((cmp = signalsToProcess.poll()) == null) && !taskInterrupted.get())  {
					Signal cmpSignal  = cmp.getStoredSignal();
					Result.Item item = new Result.Item(cmpSignal, sig.findDeviation(cmpSignal));
					result = result.include(item);
				}

				System.out.println("result done...");
				threadResultsList.put(result);
			} catch (InterruptedException e) {
				System.out.println("Processing cancelled..");
				//			e.printStackTrace();
			}
		}
	}

	private class ViewAcceptedThread implements Runnable{


		@Override
		public void run() {
			View queuedView = null;
			while (!receiveThreadFinished.get()){
				try {
					queuedView = viewAcceptedQueue.take();

					System.out.println("nodes in cluster: " + clusterSize.get());
					List<Address> members = queuedView.getMembers();
					notDegradedAnyMore = new CountDownLatch(members.size());

					degraded.set(members.size() <= 1);

					if( isNewNode.get()){
						System.out.println("GIVE ME SOME SIGNALS...");
						sendMessage(null, MessageNames.SET_ALL_DEGRADED);
						newNodeSignalsLatch = new CountDownLatch(clusterSize.get() -1);
						sendMessage(null, MessageNames.GIVE_ME_SOME_SIGNALS );
						newNodeSignalsLatch.await();
						System.out.println("EVERYBODY GAVE ME SIGNALS....");

						sendMessage(null, MessageNames.WE_ARE_ALL_OK);
					}

					Address goneNode = getGoneNode(queuedView);
					if( goneNode != null){
						executors.shutdownNow();
						taskInterrupted.set(true);
						doRedistributionProcess(goneNode);
					}

					deathSuspectedNode = null;
					isNewNode.set(false);
					setLastView(queuedView);
				} catch (InterruptedException e) {
					receiveThreadFinished.set(true);
					System.out.println("ViewAccepted Thread interrupted...");
					//	e.printStackTrace();
				}
			}
		}
	}

	private Address getGoneNode(View newView ){
		if(getLastView() == null || getLastView().getMembers().size() <= newView.getMembers().size()){
			return null;
		}else{
			for(Address node : getLastView().getMembers()){
				if( !newView.getMembers().contains(node)){
					return node;
				}
			}
		}
		return null;
	}


	private class ReceivedMessageThread implements Runnable{
		@Override
		public void run() {
			while (!receiveThreadFinished.get()){
				Message msg = null;
				try {
					msg = recievedMessagesQueue.take();

					final GenericMessage s = (GenericMessage) msg.getObject();
					switch (s.getName()) {
					case ADD_SIGNAL_TO_BACKUP:
						storeBackUpSignal(msg);
						break;
					case ADD_SIGNAL_TO_MY_LIST:
						SignalObject sig = (SignalObject)s;
						addSignalToMyList(sig.getDistributedSignal());
						break;
					case ADD_SIGNAL_TO_MY_LIST_AND_ANSWER:
						SignalObject so = (SignalObject)s;
						addSignalToMyList(so.getDistributedSignal());
						sendSignalToNode(MessageNames.REMOVE_SIGNAL_FROM_LIST, msg.getSrc(), so.getDistributedSignal() );
						break;
					case PROCESS_SIGNAL:
						System.out.println("RECIEVED PROCESS SIGNAL...");
						doProcessWork(msg);
						break;
					case RESULTS_RETURNED:
						resultsReturned(msg);
						break;
					case GIVE_ME_SOME_SIGNALS:
						giveSignalsToNode(msg.getSrc());
						break;
					case FINISHED_DISTRIBUTION_OF_SIGNALS:
						System.out.println("---- NODE " + msg.getSrc() +" FINISHED DISTRIBUTION -----");
						notDegradedAnyMore.countDown();
						//		sendMessage(Coordinator, MessageNames.NODE_FINISHED_DISTRIBUTING);
						break;
					case SET_ALL_DEGRADED:
						System.out.println("------ SETTING ALL DEGRADED -----");
						degraded.set(true);		
						break;
					case WE_ARE_ALL_OK:
						System.out.println("------ CLUSTER QUITS DEGRADATION -----");
						System.out.println("SIGNALS: " + signalsMap);
						System.out.println("BACKUPS: " + backupSignalsMap);
						System.out.println("Signals amount: " + signalsSize(signalsMap));
						System.out.println("Backups amount: "+ signalsSize(backupSignalsMap));
						degraded.set(false);
						addTaskToFindQueue();
						break;
					case REMOVE_SIGNAL_FROM_LIST:
						removeSignalFromMyList(msg);
						break;
					case MODIFY_BACKEDUP_SIGNAL:
						notifyBackUpChanged(msg);
						break;
					case NODE_FINISHED_DISTRIBUTING:
						if( redistributeLocalLatch != null)
							redistributeLocalLatch.countDown();
						break;
					case SEND_SIGNALS_TO_NODE:
						SignalListObject signals = (SignalListObject)s;
						addSignalsFromNode( msg.getSrc(), signals);
						break;
					case SEND_SIGNALS_TO_NODES_BACKUP:
						SignalListObject backupSignals = (SignalListObject)s;
						addSignalsFromNode( msg.getSrc(), backupSignals);
						break;
					case SEND_SIGNALS_TO_NEW_NODE:
						SignalListObject newSignals = (SignalListObject)s;
						recievedSignalsFromNode(msg.getSrc(), newSignals);
					case SIGNALS_SENT_TO_NODE:
						//		newNodeSignalsLatch.countDown();
					default:
						break;
					}		
				} catch (InterruptedException e) {
					receiveThreadFinished.set(true);
					System.out.println("Recieve Message Thread interrupted...");
					//	e.printStackTrace();
				}
			}
		}
	}

	private synchronized void recievedSignalsFromNode(Address src, SignalListObject newSignals) {
		if( !newSignals.getSignals().isEmpty()){
			for( DistributedSignal s : newSignals.getSignals() ){
				if( signalsMap.get(s.getWhoBacksItUp()) == null){
					List<DistributedSignal> l = new LinkedList<>();
					l.add(s);
					signalsMap.put(s.getWhoBacksItUp(), l);
				}else{
					signalsMap.get(s.getWhoBacksItUp()).add(s);
				}
			}
			System.out.println("RECEIVED SIGNALS FROM NODE: "+src);
		}
		newNodeSignalsLatch.countDown();

		//sendMessage(src, MessageNames.FINISED_NEW_NODE_DISTRIBUTION);
	}


	private synchronized void addSignalsFromNode(Address src, SignalListObject signals) {
		if( signals.getSignals() != null && !signals.getSignals().isEmpty()){
			if( signals.getName() == MessageNames.SEND_SIGNALS_TO_NODE){
				for( DistributedSignal s: signals.getSignals()){
					Address whoBacksItUp = s.getWhoBacksItUp();
					if( signalsMap.get(whoBacksItUp) == null){
						LinkedList<DistributedSignal> sigs = new LinkedList<>();
						sigs.add(s);
						signalsMap.put(whoBacksItUp,sigs);
					}else{
						signalsMap.get(whoBacksItUp).add(s);
					}
				}
			
				System.out.println("I GOT YOUR SIGNALS...");
			}else if (signals.getName() == MessageNames.SEND_SIGNALS_TO_NODES_BACKUP){
				for( DistributedSignal s: signals.getSignals()){
					Address whoHasIt = s.getWhoHasIt();
					if( backupSignalsMap.get(whoHasIt) == null){
						LinkedList<DistributedSignal> sigs = new LinkedList<>();
						sigs.add(s);
						backupSignalsMap.put(whoHasIt, sigs);
					}else{
						backupSignalsMap.get(whoHasIt).add(s);
					}
				}
				
			}
			sendMessage(src, MessageNames.NODE_FINISHED_DISTRIBUTING);
		}
	}




	public void resultsReturned(Message msg){
		System.out.println("i recieved results from node: " +msg.getSrc());
		ResultObject r = (ResultObject) msg.getObject();

		try {
			resultsQueue.put(r.getResult());
			countLatch.countDown();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}
